package com.shujia.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo1WordCount {
  def main(args: Array[String]): Unit = {
    /**
     * spark core: SparkContext  --> 核心数据类型 RDD
     * spark sql: SparkSession --> 核心数据类型 DataFrame
     * spark streaming: StreamingContext --> 核心数据类型 DStream(RDD)
     *
     * 创建SparkStreaming的环境需要使用StreamingContext对象
      */
    //spark streaming依赖于spark core的环境
    //因为spark streaming核心数据类型底层封装的是RDD
    val conf = new SparkConf()
    conf.setMaster("local[2]")
    conf.setAppName("spark streaming 单词统计案例1")
    val sparkContext = new SparkContext(conf)
    // def this(sparkContext: SparkContext, batchDuration: Duration) //传入一个spark core 环境对象以及一个接收时间的范围
    val streamingContext = new StreamingContext(sparkContext, Durations.seconds(5))

    /**
     * 因为sparkstreaming是一个近实时的计算引擎，整个程序写完运行的状态，一直运行，一直接收数据，除非异常或者手动停止
     *
     * 1、目前在spark使用nc工具监控一个端口号中源源不断产生的数据
     * 2、yum install -y nc 下载安装nc工具
     * 3、nc -lk xxxx
     */
    val lineDS: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 12345)

    val flatMapDS: DStream[String] = lineDS.flatMap(_.split(" "))
    val wordsKVDS: DStream[(String, Int)] = flatMapDS.map((_, 1))
    val resultDS: DStream[(String, Int)] = wordsKVDS.reduceByKey(_ + _)

    //后续的操作和在spark core中计算单词统计的方式一致
//    val resultDS: DStream[(String, Int)] = lineDS.flatMap(_.split(" "))
//      .map((_, 1))
//      .reduceByKey(_ + _)

    //打印结果
    resultDS.print()

    /**
     * Spark Streaming作业触发执行的方式和之前完全不一样
     */

    streamingContext.start()
    streamingContext.awaitTermination()
    streamingContext.stop()
  }
}
